﻿using System;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using Jiubang.Communication.Core;
using Jiubang.Communication.Server.Buffers;
using Jiubang.Communication.Server.RequestRouter;
using Jiubang.Util;
using log4net;

namespace Jiubang.Communication.Server.DataTransport {
    /// <summary>
    ///   基于APM实现Socket服务端连接和数据接收、发送逻辑。
    ///   支持长连接，每个消息片断用2bytes uint16定义正文长度。
    ///   Receive方法绑定一个SocketAsyncEventArgs；Send方法SocketAsyncEventArgs实例由对象池管理<br />
    ///   一次完整处理异步步骤如下：<br />
    ///   1.AcceptComplete 接受并注册新Socket连接，开始第一条消息长度读取<br />
    ///   2.IoCompleted -> ReceiveMessageHeader 获取消息长度，开始正文读取<br />
    ///   3.IoCompleted -> ReceiveMessageBody 获取消息正文，调用Dispatcher.Accept<br />
    ///   4.Send被Dispatcher类调用<br />
    ///   5.IoCompleted -> SendMessage<br />
    /// </summary>
    public sealed class SocketServer : IDisposable, IMessageSender, IMessageListener {
        internal static readonly int messageHeadLength = 2;
        private readonly Buffers.BufferManager _bufferManager;
        private readonly WaitCallback _ioCompletedHandler;
        private readonly Socket _listenerSocket;
        private readonly ILog _log;
        private readonly PerformanceCounter _prefCounter;
        private readonly SocketAsyncEventArgs _socketAcceptedEventArgs;
        private readonly WaitCallback _socketAcceptedHandler;
        private readonly ObjectPool<SocketAsyncEventArgs> _socketAsyncEventArgsPool;

        private bool _disposed;


        /// <summary>
        ///   初始化Socket服务环境并开始接受Socket连接
        /// </summary>
        /// <param name = "instanceName">实例名称，用于性能监控。同一服务器内不能重复。</param>
        /// <param name = "localEndPoint">当前实例监听的EndPoint</param>
        /// <param name = "bufferManager">接收消息时将向该对象申请缓冲区</param>
        public SocketServer(string instanceName, EndPoint localEndPoint, Buffers.BufferManager bufferManager) {
            if (bufferManager == null)
                throw new ArgumentNullException("bufferManager");

            _listenerSocket = new Socket(localEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
            _listenerSocket.Bind(localEndPoint);

            _prefCounter = new PerformanceCounter(instanceName);

            _socketAcceptedEventArgs = new SocketAsyncEventArgs();
            _socketAcceptedEventArgs.Completed += AcceptCompleted;

            _socketAsyncEventArgsPool = new ObjectPool<SocketAsyncEventArgs>(1024, () => {
                var args = new SocketAsyncEventArgs();
                args.Completed += IOCompleted;
                return args;
            });

            _socketAcceptedHandler = o => AcceptCompleted(this, (SocketAsyncEventArgs)o);
            _ioCompletedHandler = o => IOCompleted(this, (SocketAsyncEventArgs)o);

            _disposed = false;
            _log = LogManager.GetLogger(GetType());
            _bufferManager = bufferManager;

            _listenerSocket.Listen(int.MaxValue);
        }

        #region IDisposable Members

        /// <summary>
        ///   设置为Dispose状态并回收ListenerSocket
        /// </summary>
        public void Dispose() {
            _disposed = true;
            _listenerSocket.Close();
            _prefCounter.Dispose();
        }

        #endregion

        #region public members

        /// <summary>
        ///   接收到完整的消息内容触发的事件。
        ///   SocketAsyncEventArgs在事件回调执行完毕后将被回收。
        ///   回收时如SocketAsyncEventArgs.Buffer不为null，
        ///   则Buffer也将被BufferManager回收。
        ///   如需要自行管理Buffer回收及重用，
        ///   需要在回调方法中将Buffer属性设置为null。
        /// </summary>
        public event EventHandler<SocketAsyncEventArgs> MessageReceived;

        /// <summary>
        ///   接受一个新的Socket并获取SocketAsyncEventArgs对象用于接收消息时触发的事件。
        ///   在该事件回调可以通过UserToken标识是否初次数据接收。
        /// </summary>
        internal event EventHandler<SocketAsyncEventArgs> SocketAccepted;

        /// <summary>
        ///   Socket关闭时触发的事件
        /// </summary>
        internal event EventHandler<SocketAsyncEventArgs> SocketClosed;

        /// <summary>
        ///  消息发送完毕或由Server克隆缓冲区后的回调
        /// </summary>
        public event EventHandler<SocketAsyncEventArgs> SendBufferReleased;

        /// <summary>
        /// 使服务处于监听状态
        /// </summary>
        /// <param name="backlog"></param>
        public void Listen(int backlog) {
            _listenerSocket.Listen(backlog);
        }

        /// <summary>
        ///   发送消息头及正文
        /// </summary>
        /// <param name = "socket">发送目标</param>
        /// <param name = "buffer">发送的消息正文缓冲区。</param>
        /// <returns>是否异步发送</returns>
        public bool SendAsync(Socket socket, ArraySegment<byte> buffer) {
            return SendAsync(socket, buffer, null);
        }

        /// <summary>
        ///   发送消息头及正文
        /// </summary>
        /// <param name = "socket">发送目标</param>
        /// <param name = "buffer">发送的消息正文缓冲区。</param>
        /// <param name="bufferCounter">缓冲区共享计数器</param>
        /// <returns>是否异步发送</returns>
        internal bool SendAsync(Socket socket, ArraySegment<byte> buffer, ShareBufferCounter bufferCounter) {
            if (buffer.Offset != messageHeadLength) {
                throw new Exception("发送缓冲区的起始位偏移量不为" + messageHeadLength);
            }

            var args = _socketAsyncEventArgsPool.Get();
            args.AcceptSocket = socket;
            args.SetBuffer(buffer.Array, 0, buffer.Count - messageHeadLength);
            args.UserToken = bufferCounter;


            var bufferLength = (ushort)(args.Count - 2);
            args.Buffer[0] = (byte)bufferLength;
            args.Buffer[1] = (byte)(bufferLength >> 8);


            if (!args.AcceptSocket.SendAsync(args)) {
                IOCompleted(this, args);
                return false;
            }
            return true;
        }

        #endregion

        #region primary event handlers

        /// <summary>
        ///   事件回调执行以下逻辑：
        ///   1.检查Server是否关闭；
        ///   2.申请SocketAsyncEventArgs并关联Accept的Socket；
        ///   3.触发SocketAccepted事件；
        ///   4.开始该Socket的异步接收；
        ///   5.开始Accept另一个Socket。
        /// </summary>
        /// <param name = "sender"></param>
        /// <param name = "args"></param>
        private void AcceptCompleted(object sender, SocketAsyncEventArgs args) {
            try {
                if (_disposed) {
                    CloseSocket(args);
                    return;
                }

                _prefCounter.Accept();
                ReceiveMessageHeaderAsync(args.AcceptSocket, SocketAccepted);

                //重用SocketAsyncEventArgs接收新Socket
                args.AcceptSocket = null;
                if (!_listenerSocket.AcceptAsync(args)) {
                    //避免递归造成栈溢出，这里性能基本不成问题
                    ThreadPool.QueueUserWorkItem(_socketAcceptedHandler, args);
                }
            } catch (Exception ex) {
                _log.Error(ex);
            }
        }


        /// <summary>
        ///   处理IO完成事件。<br />
        ///   如发现异常立即断开Socket。客户端重连后将创建新的Socket。<br />
        /// </summary>
        private void IOCompleted(object sender, SocketAsyncEventArgs args) {
            if (_disposed) {
                CloseSocket(args);
                return;
            }

            try {
                if (args.SocketError != SocketError.Success || args.BytesTransferred <= 0) {
                    _log.WarnFormat("{0}操作结果异常[{1}]，传输长度为{2}。socket[{3}]:{4}", args.LastOperation, args.SocketError,
                                    args.BytesTransferred, GetEndPointDescription(args.AcceptSocket.RemoteEndPoint),
                                    args.Buffer.Concat(","));
                    CloseSocket(args);
                    return;
                }

                switch (args.LastOperation) {
                    case SocketAsyncOperation.Receive:
                        ReceiveCompleted(args);
                        break;
                    case SocketAsyncOperation.Send:
                        _prefCounter.Send(args.Count);

                        var buffer = args.UserToken as byte[];
                        if (buffer != null) {
                            _bufferManager.Return(args.Buffer);
                        } else if (SendBufferReleased != null) {
                            SendBufferReleased(this, args);
                        }
                        ReleaseEventArgs(args);
                        break;
                    default:
                        throw new InvalidOperationException("未被支持的操作" + args.LastOperation);
                }
            } catch (Exception e) {
                _log.Error(e);
                CloseSocket(args);
            }
        }

        #endregion

        #region receiving logics

        /// <summary>
        ///   递归读取消息直到预置缓冲区填充完毕。
        ///   当读取消息头时起始缓冲区大小为messageHeadLength；
        ///   读取正文时缓冲区大小为消息头声明长度。
        ///   数据读取后缓冲区将缩小至剩余消息长度。
        ///   <br />
        ///   根据args.Offset属性是否0判断是否完成消息正文获取。
        /// </summary>
        private void ReceiveCompleted(SocketAsyncEventArgs args) {
            if (args.BytesTransferred > args.Count)
                throw new Exception(string.Format("实际接收长度【{0}】,期望长度【{1}】", args.BytesTransferred, args.Count));

            if (args.BytesTransferred < args.Count) {
                args.SetBuffer(args.Offset + args.BytesTransferred, args.Count - args.BytesTransferred);

                if (_log.IsDebugEnabled) {
                    _log.DebugFormat("消息读取内容剩余{0}byte(s)\r\n消息来自{1}", (args.Count),
                                     GetEndPointDescription(args.AcceptSocket.RemoteEndPoint));
                }

                if (!args.AcceptSocket.ReceiveAsync(args))
                    IOCompleted(this, args);
                return;
            }


            if (args.Offset == 0) {
                ReceiveMessageBody(args);
            } else {
                if (MessageReceived == null)
                    throw new NullReferenceException("MessageReceived事件未注册");

                _prefCounter.Receive(args.Buffer.Length);

                ReceiveMessageHeaderAsync(args.AcceptSocket, null);

                args.SetBuffer(messageHeadLength, args.Buffer.Length - messageHeadLength);

                try {
                    MessageReceived(this, args);
                } catch (Exception e) {
                    _log.Warn("执行以下命令时发生异常[" + args.Buffer.Concat(",") + "]", e);
                }

                _bufferManager.Return(args.Buffer);
                ReleaseEventArgs(args);
            }
        }

        private void ReceiveMessageHeaderAsync(Socket socket, EventHandler<SocketAsyncEventArgs> eventHandler) {
            var socketAsyncEventArgs = _socketAsyncEventArgsPool.Get();
            socketAsyncEventArgs.AcceptSocket = socket;
            socketAsyncEventArgs.SetBuffer(_bufferManager.Apply(messageHeadLength), 0, messageHeadLength);

            if (eventHandler != null)
                eventHandler(this, socketAsyncEventArgs);

            if (!socket.ReceiveAsync(socketAsyncEventArgs)) {
                //避免长期进入递归状态
                ThreadPool.QueueUserWorkItem(_ioCompletedHandler, socketAsyncEventArgs);
            }
        }

        private void ReceiveMessageBody(SocketAsyncEventArgs args) {
            var appliedBufferLength = (ushort)(BitConverter.ToUInt16(args.Buffer, 0) + 2);

            //判断缓冲区长度足够)
            if (args.Count >= appliedBufferLength) {
                args.SetBuffer(args.Offset + args.BytesTransferred, appliedBufferLength - 2);
            } else {
                //缓冲区长度不足从锯齿缓冲区申请
                var buffer = _bufferManager.Apply(appliedBufferLength);

                //复制已读消息至新缓冲区
                Buffer.BlockCopy(args.Buffer, 0, buffer, 0, 2);

                //回收缓冲区空间
                _bufferManager.Return(args.Buffer);

                //设置新缓冲区
                args.SetBuffer(buffer, 2, appliedBufferLength - 2);
            }

            if (!args.AcceptSocket.ReceiveAsync(args))
                IOCompleted(this, args);
        }

        #endregion

        #region helper methods

        /// <summary>
        ///   清空SocketAsyncEventArgs对象引用资源，并尝试回收到对象池。
        ///   如池已满则执行Dispose后释放。
        /// </summary>
        /// <param name = "args"></param>
        private void ReleaseEventArgs(SocketAsyncEventArgs args) {
            args.SetBuffer(null, 0, 0);
            args.AcceptSocket = null;
            args.UserToken = null;
            if (!_socketAsyncEventArgsPool.Set(args))
                args.Dispose();
        }


        private static string GetEndPointDescription(EndPoint endPoint) {
            var endpoint = endPoint as IPEndPoint;
            if (endpoint != null) return endpoint.Address + ":" + endpoint.Port;
            return endPoint.ToString();
        }

        /// <summary>
        ///   1.触发SocketClosed事件；
        ///   2.主动关闭Socket；
        ///   3.回收SocketAsyncEventArgs对象。
        ///   <br />
        ///   由于无法确认缓冲区是否仍然有其它对象引用，
        ///   清空但不回收缓冲区。
        ///   <br />
        ///   内部逻辑自带异常捕获逻辑，
        ///   调用程序不需要加异常处理。
        /// </summary>
        /// <param name = "eventArgs"></param>
        private void CloseSocket(SocketAsyncEventArgs eventArgs) {
            if (SocketClosed != null) {
                try {
                    SocketClosed(this, eventArgs);
                } catch (Exception e) {
                    _log.Warn(e);
                }
            }

            try {
                _prefCounter.Close();

                var socket = eventArgs.AcceptSocket;
                if (socket != null) {
                    if (socket.Connected) {
                        socket.Shutdown(SocketShutdown.Both);
                        _log.Info("Socket[" + GetEndPointDescription(socket.RemoteEndPoint) + "]已经主动关闭");
                    }
                    socket.Close();
                }

                ReleaseEventArgs(eventArgs);
            } catch (Exception e) {
                _log.Error(e);
            }
        }

        #endregion
    }
}